package com.xrui.hbasecmd.hbase;

import com.google.common.base.Preconditions;
import com.google.common.collect.Iterables;
import com.xrui.hbase.*;
import com.xrui.hbase.conversion.CellConverter;
import com.xrui.hbase.model.FieldKey;
import com.xrui.hbase.model.RecordFields;
import com.xrui.hbase.schema.TableLayout;
import com.xrui.hbasecmd.config.GlobalConfig;
import com.xrui.hbasecmd.data.*;
import com.xrui.hbasecmd.export.FileExporter;
import com.xrui.hbase.util.ResourceUtils;
import org.apache.commons.lang.time.StopWatch;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.IOException;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.util.*;

/**
 * This class represents a scanner over the hbase tables.
 */
@SuppressWarnings({"OverlyNestedMethod", "ClassWithTooManyMethods"})
public class QueryScanner implements Closeable {
    //region Constants
    private final static Logger LOG = LoggerFactory.getLogger(QueryScanner.class);
    //endregion
    //region Variables
    /**
     * A constant representing a size of the unicode character.
     */
    private static final int UNICODE_CHAR_SIZE = 4;
    /**
     * The connection which owns the scanner.
     */
    protected Connection connection;
    /**
     * The name of the hbase table the scanner is going to be executed on.
     */
    protected String tableName;
    private ColumnName columnName;
    private CellConverter cellConverter;
    /**
     * A query to be used with he scanner.
     */
    private Query query;
    /**
     * Holds a total number of rows in the table. This parameter is calculated by scanning over the whole table.
     * This operation is time consuming and its result should be cached.
     */
    private long rowsCount;
    /**
     * Holds a rows number in the table if the calculation has not been completed because of the timeout.
     */
    private long partialRowsCount;
    /**
     * The number of the last loaded row.
     */
    private long lastRow;
    /**
     * The map of column types. The key is the name of the column and the value is the type of the objects within the column.
     */
    private Map<String, ColumnType> columnTypes;
    /**
     * A list of markers. The marker is used for pagination to mark where the previous scan has stopped.
     */
    private Stack<Marker> markers;
    /**
     * A list of loaded rows.
     */
    private Collection<DataRow> current;
    /**
     * Indicates if this scanner should support only forward navigation.
     */
    private boolean forwardNavigateOnly;

    /**
     * Initializes a new instance of the {@link QueryScanner} class.
     *
     * @param connection The reference to the connection.
     * @param tableName  The name of the table to be scanned.
     */
    public QueryScanner(Connection connection, String tableName, ColumnName columnName) throws IOException {
        this.connection = connection;
        this.tableName = tableName;
        this.rowsCount = 0;
        this.lastRow = 0;
        this.markers = new Stack<>();
        this.columnName = columnName;
        this.cellConverter = connection.getTableLayout(tableName).getConverter(columnName);
    }

    /**
     * Initializes a new instance of the {@link QueryScanner} class.
     *
     * @param connection The reference to the connection.
     * @param tableName  The name of the table the scan is going to be done on.
     * @param query      The query to use.
     */
    public QueryScanner(Connection connection, String tableName, Query query) throws IOException {
        this(connection, tableName, ColumnName.create(query.getColumnWithFamily()));
        this.query = query;
    }
    //endregion

    //region Constructor

    /**
     * Adds additional 4 bytes to the end key byte array. This prevents the scan go further than the end key.
     *
     * @param endKey The end key to update.
     * @return A new byte array containing extra bytes.
     */
    public static byte[] padWithMaxUnicode(byte[] endKey) {
        ByteBuffer buffer = ByteBuffer.allocate(endKey.length + UNICODE_CHAR_SIZE);
        buffer.put(endKey);
        buffer.put(Byte.MAX_VALUE);
        buffer.put(Byte.MAX_VALUE);
        buffer.put(Byte.MAX_VALUE);
        buffer.put(Byte.MAX_VALUE);
        return buffer.array();
    }
    //endregion

    //region Public Properties

    public void setColumnName(ColumnName columnName) {
        this.columnName = columnName;
    }

    /**
     * Sets a new query to use.
     *
     * @param query A new query to set.
     */
    @SuppressWarnings("ObjectEquality")
    public void setQuery(Query query) {
        if (this.query != query) {
            resetCurrent();
        }
        this.query = query;
    }

    /**
     * Gets a reference to the {@link Connection}.
     *
     * @return A reference to the {@link Connection}.
     */
    public Connection getConnection() {
        return this.connection;
    }

    /**
     * Gets a reference to the {@link Configuration} instance used by connection.
     *
     * @return A reference to the {@link Configuration} instance.
     */
    public Configuration getConfiguration() {
        return this.connection.getConfiguration();
    }

    /**
     * Gets the value indicating if this scanner supports only forward navigation.
     *
     * @return True if this instance of the scanner supports only forward navigation or False if both forward and backward navigation are supported.
     */
    public boolean getForwardNavigateOnly() {
        return this.forwardNavigateOnly;
    }

    /**
     * Sets the value indicating if this scanner should support forward navigate only.
     *
     * @param forwardNavigateOnly True if this instance of the scanner should support only forward navigation or False to support both forward and backward navigations.
     */
    public void setForwardNavigateOnly(boolean forwardNavigateOnly) {
        this.forwardNavigateOnly = forwardNavigateOnly;
    }

    /**
     * Gets the name of the table.
     *
     * @return The table name.
     */
    public String getTableName() {
        return this.tableName;
    }

    /**
     * Gets the name of the column.
     *
     * @return The column name.
     */
    public ColumnName getColumnName() {
        return this.columnName;
    }

    /**
     * Gets the cell converter.
     *
     * @return The cell converter.
     */
    public CellConverter getCellConverter() {
        return this.cellConverter;
    }

    public RowScanner newScanner() throws IOException {
        Query query = getQuery();
        return getRowScanner(query);
    }

    /**
     * Indicates that the rows count has been partially calculated.
     *
     * @return True if the rows count has been stopped before reaching end of the table or False otherwise.
     */
    public boolean isRowsCountPartiallyCalculated() {
        return this.partialRowsCount > 0;
    }

    /**
     * Gets a list of columns. If there is no columns at this moment they will be loaded according to
     * the provided rows number. In other words only columns of loaded rows will be returned.
     *
     * @return A list of columns.
     */
    public Collection<TypedColumn> getColumns() {
        if (this.markers.isEmpty()) {
            return null;
        } else {
            return peekMarker().columns;
        }
    }

    /**
     * Gets a mapping of column names to column types.
     *
     * @return A map of column types.
     */
    public Map<String, ColumnType> getColumnTypes() {
        return this.columnTypes;
    }

    /**
     * Sets a mapping of column names to column types.
     *
     * @param columnTypes A new map of column types.
     */
    public void setColumnTypes(Map<String, ColumnType> columnTypes) {
        this.columnTypes = columnTypes;
    }

    /**
     * Gets the last loaded row number.
     *
     * @return The number of the last loaded row.
     */
    public long getLastRow() {
        return this.lastRow;
    }

    /**
     * Indicates if there is more rows to show.
     *
     * @return True if there is more rows to show or False otherwise.
     */
    public boolean hasNext() {
        return !markers.isEmpty() && peekMarker().hasNext;
    }
    //endregion

    //region Public Methods

    /**
     * Indicates if the backward navigation is possible.
     *
     * @return True if there are previously loaded rows or False otherwise.
     */
    public boolean hasPrev() {
        return this.markers.size() > 1;
    }

    /**
     * Validates whether the values held by the column can be converted to the specified type.
     *
     * @param columnType The type to check.
     * @return True if the value can be converted to the specified type or False otherwise.
     */
    public boolean isColumnOfType(String columnName, ColumnType columnType) {
        Collection<DataRow> rows = this.current;
        if (rows != null) {
            int counter = 0;
            Iterator<DataRow> iterator = rows.iterator();

            while (counter < 10 && iterator.hasNext()) {
                DataRow row = iterator.next();
                if (!row.isCellOfType(columnName, columnType)) {
                    return false;
                }
                counter++;
            }

            return true;
        }
        return true;
    }

    /**
     * Updates a column type of all cells in the specified column.
     *
     * @param columnName The name of the column.
     * @param columnType The new column type.
     */
    public void updateColumnType(String columnName, ColumnType columnType) {
        Collection<DataRow> rows = this.current;
        if (rows != null) {
            for (DataRow row : rows) {
                row.updateColumnType(columnName, columnType);
            }
        }
    }

    /**
     * Resets the cache.
     */
    public void resetCurrent() {
        this.current = null;
        this.rowsCount = 0;
        this.lastRow = 0;
        this.markers.clear();

    }

    /**
     * Gets a first non empty row or the first row in the table.
     *
     * @return A first non empty row or the first row in the table.
     * @throws IOException Error accessing hbase.
     */
    public DataRow getFirstNonEmptyRowOrFirstRow(int endAtMost) throws IOException {
        try (RowScanner rowScanner = getRowScanner()) {
            Collection<DataRow> rows = new LinkedList<>();
            loadRows(rowScanner, 0, endAtMost, rows);

            Optional<DataRow> firstNonEmptyRow = rows.stream().filter(it -> Iterables.size(it.getCells()) > 3).findFirst();

            return firstNonEmptyRow.orElse(rows.stream().findFirst().orElse(null));
        }
    }

    /**
     * Gets a first row in the table.
     *
     * @return A first row in the table.
     * @throws IOException Error accessing hbase.
     */
    public DataRow getFirstRow() throws IOException {
        try (RowScanner rowScanner = getRowScanner()) {
            Collection<DataRow> rows = new LinkedList<>();

            loadRows(rowScanner, 0, 1, rows);
            if (rows.isEmpty()) {
                return null;
            }
            return rows.iterator().next();
        }
    }

    /**
     * Gets a list of already loaded rows.
     *
     * @return A list of rows.
     */
    public Collection<DataRow> current() {
        return this.current;
    }

    /**
     * Gets a list of rows loaded starting from the beginning.
     *
     * @param rowsNumber The number of rows to load.
     * @return A list of rows.
     * @throws IOException Error accessing hbase.
     */
    public Collection<DataRow> current(int rowsNumber) throws IOException {
        return current(0, rowsNumber);
    }

    /**
     * Gets a list of rows loaded starting from the offset.
     *
     * @param offset     The first row to start loading from.
     * @param rowsNumber The number of rows to load.
     * @return A list of rows.
     * @throws IOException Error accessing hbase.
     */
    public Collection<DataRow> current(long offset, int rowsNumber) throws IOException {
        if (this.current == null ||
            this.current.size() != rowsNumber ||
            this.current.size() + offset != this.lastRow) {

            // offset should start from 1.
            if (offset == 0) {
                offset++;
            }

            this.markers.clear();
            this.current = next(offset - 1, rowsNumber);
            this.lastRow = offset + this.current.size() - 1;
        }
        return this.current;
    }

    /**
     * Gets a list of rows loaded starting from the previous position.
     *
     * @param rowsNumber The number of rows to load.
     * @return A list of rows.
     * @throws IOException Error accessing hbase.
     */
    public Collection<DataRow> next(int rowsNumber) throws IOException {
        int offset = this.markers.isEmpty() ? 0 : this.markers.peek().offset;
        this.current = next(offset, rowsNumber);
        this.lastRow += this.current.size();

        return this.current;
    }

    /**
     * Gets a list of previously loaded rows.
     *
     * @return A list of rows.
     * @throws IOException Error accessing hbase.
     */
    public Collection<DataRow> prev() throws IOException {
        if (!this.markers.isEmpty()) {
            if (this.markers.size() > 1) {
                popMarker();
            }

            this.lastRow -= this.current.size();
            this.current = peekMarker().rows;

            for (Map.Entry<String, ColumnType> entry : this.columnTypes.entrySet()) {
                updateColumnType(entry.getKey(), entry.getValue());
            }
        }
        return this.current;
    }
    //endregion

    //region Protected Methods

    /**
     * Gets the total number of rows in the table. This value is calculated by scanning throughout the whole table to count the rows. This value is then cached
     * for future uses.
     *
     * @return A total number of rows in the table.
     * @throws IOException Error accessing hbase.
     */
    public long getRowsCount(long timeout) throws IOException {
        if (this.rowsCount == 0) {
            this.partialRowsCount = 0;
            boolean hasTimeout = false;

            StopWatch stopWatch = new StopWatch();
            stopWatch.start();
            int count = 0;
            RowScanner rowScanner = null;
            try {
                rowScanner = getRowScanner();
                while (!hasTimeout && rowScanner.iterator().hasNext()) {
                    RowData rowData = rowScanner.iterator().next();

                    for (HBaseCell<Object> ignored : rowData.asIterable(columnName.getFamily())) {
                        count++;
                        if (stopWatch.getTime() > timeout) {
                            this.partialRowsCount = count;
                            hasTimeout = true;
                            break;
                        }
                    }

                }

                this.rowsCount = count;
            } finally {
                ResourceUtils.closeOrLog(rowScanner);
                stopWatch.stop();
            }
        }
        return this.rowsCount;
    }

    public DataRow fulfilDataRow(
        RowData rowData, HBaseCell<Object> cell, CellConverter cellConverter
    ) throws IOException {
        byte[] rowKey = rowData.getEntityId().getHBaseRowKey();
        RecordFields recordFields = Preconditions.checkNotNull(cellConverter).convert(cell);
        DataRow row = new DataRow(new ConvertibleObject(ColumnType.BinaryString, rowKey));
        for (Map.Entry<FieldKey<?>, Object> field : recordFields.entrySet()) {
            FieldKey fieldKey = field.getKey();
            Object value = field.getValue() == null ? "" : field.getValue();

            String typeName = fieldKey.getType() == null ? "String" : fieldKey.getType().getSimpleName();
            ColumnType columnType = ColumnType.fromNameOrDefault(typeName, ColumnType.String);
            if (columnType.equals(ColumnType.String)) {
                columnType = ConvertibleObject.guessType(value);
            }

            TypedColumn column = new TypedColumn(fieldKey.getName(), columnType);
            ConvertibleObject convertibleValue = new ConvertibleObject(columnType, toBytes(value));
            row.addCell(new DataCell(row, column, convertibleValue));
        }

        return row;
    }

    /**
     * Export the rows.
     *
     * @param exporter The exporter to export the records.
     * @throws IOException Error accessing hbase or writing file.
     */
    public void export(FileExporter exporter) throws IOException {
        try (RowScanner scanner = newScanner()) {
            while (scanner.iterator().hasNext()) {
                RowData rowData = scanner.iterator().next();
                for (HBaseCell<Object> cell : rowData.asIterable(columnName.getFamily())) {
                    DataRow row = fulfilDataRow(rowData, cell, cellConverter);
                    exporter.write(row);
                }
            }
        }
    }

    /**
     * Gets the hbase row scanner.
     *
     * @return A hbase row scanner
     * @throws IOException Error accessing hbase.
     */
    private RowScanner getRowScanner() throws IOException {
        return getRowScanner(this.query);
    }

    /**
     * Gets an hbase row scanner.
     *
     * @return An hbase row scanner.
     * @throws IOException Error accessing hbase.
     */
    private RowScanner getRowScanner(Query query) throws IOException {
        DataRequestBuilder.ColumnsDef columnsDef = DataRequestBuilder.ColumnsDef.create();
        TableLayout tableLayout = connection.getTableLayout(tableName);
        DataRequestBuilder dataRequestBuilder = DataRequest.builder();

        TableReader.ScannerOptions scannerOptions = new TableReader.ScannerOptions();

        if (query != null) {
            if (query.getMaxVersions() > 1) {
                columnsDef.withMaxVersions(query.getMaxVersions());
            }
            if (query.getStartKey() != null) {
                scannerOptions.setStartRow(HBaseEntityId.create(query.getStartKey(), tableLayout));
            }

            if (query.getEndKey() != null) {
                scannerOptions.setStopRow(HBaseEntityId.create(query.getEndKey(), tableLayout));
            }

            if (query.getRowFilter() != null) {
                scannerOptions.setRowFilter(query.getRowFilter());
            }

            if (query.getStartDate() != null && query.getEndDate() != null) {
                dataRequestBuilder.withTimeRange(query.getStartDate().getTime(), query.getEndDate().getTime());
            }
        }

        columnsDef.add(columnName);
        dataRequestBuilder.addColumns(columnsDef);
        DataRequest dataRequest = dataRequestBuilder.build();
        Table table = null;
        TableReader tableReader = null;
        RowScanner scanner;
        try {
            table = connection.getTable(tableName);
            tableReader = table.openTableReader();
            scanner = tableReader.getScanner(dataRequest, scannerOptions);
        } finally {
            ResourceUtils.closeOrLog(tableReader);
            ResourceUtils.releaseOrLog(table);
        }
        return scanner;
    }

    //endregion

    //region Private Methods
    protected byte[] toBytes(Object value) {
        if (null != value) {
            if (value instanceof Long) {
                return Bytes.toBytes((Long) value);
            } else if (value instanceof Short) {
                return Bytes.toBytes((Short) value);
            } else if (value instanceof Double) {
                return Bytes.toBytes((Double) value);
            } else if (value instanceof Float) {
                return Bytes.toBytes((Float) value);
            } else if (value instanceof Boolean) {
                return Bytes.toBytes((Boolean) value);
            } else if (value instanceof Integer) {
                return Bytes.toBytes((Integer) value);
            } else if (value instanceof BigDecimal) {
                return Bytes.toBytes((BigDecimal) value);
            } else if (value instanceof byte[]) {
                return (byte[]) value;
            } else {
                return Bytes.toBytes(String.valueOf(value));
            }
        }
        return new byte[0];
    }

    private Query getQuery() {
        Query query;
        if (this.query != null) {
            query = this.query;
        } else {
            query = new Query();
        }

        if (!this.markers.isEmpty()) {
            query.setStartKey(peekMarker().endKey);
        }
        return query;
    }

    /**
     * Loads a specified number of rows from the hbase.
     *
     * @param scanner    The hbase scanner to retrieve the data.
     * @param offset     The offset to start from.
     * @param rowsNumber The number of rows to load.
     * @param rows       The loaded rows. This is the output parameter.
     * @return A key of the last loaded row. Used to mark the current position for the next scan.
     * @throws IOException Error accessing hbase.
     */
    private Marker loadRows(
        RowScanner scanner, long offset, int rowsNumber, Collection<DataRow> rows
    ) throws IOException {
        int i = 0;
        int j = 0;
        boolean hasNext = false;
        boolean fullPopulated = false;
        byte[] startKey = null;
        byte[] endKey = null;

        // Scan 1 more row to identify whether has next row
        while (scanner.iterator().hasNext() && rows.size() < rowsNumber + 1 && !hasNext) {
            RowData rowData = scanner.iterator().next();
            if (rows.size() < rowsNumber) {
                j = 0;
            }

            for (HBaseCell<Object> cell : rowData.asIterable(columnName.getFamily())) {
                if (fullPopulated) {
                    LOG.debug("At least one or more row(s) found after got {} rows", rows.size());
                    hasNext = true;
                    break;
                } else if (rows.size() == rowsNumber) {
                    fullPopulated = true;
                    continue;
                }

                if (i >= offset) {
                    byte[] rowKey = rowData.getEntityId().getHBaseRowKey();

                    if (startKey == null) {
                        startKey = rowKey;
                    }
                    endKey = rowKey;

                    DataRow row = fulfilDataRow(rowData, cell, cellConverter);
                    rows.add(row);
                }
                j++;
                i++;
            }
        }

        // fulfil columns with last row
        Collection<TypedColumn> columns = new LinkedList<>();
        if (!Iterables.isEmpty(rows)) {
            DataRow lastRow = Iterables.getLast(rows);
            for (DataCell cell : lastRow.getCells()) {
                columns.add(cell.getColumn());
            }
        }

        LOG.info("Got {} rows", rows.size());
        return new Marker(startKey, endKey, j, hasNext, rows, columns);
    }

    /**
     * Loads the following rows.
     *
     * @param offset     The offset to start from.
     * @param rowsNumber The number of rows to load.
     * @return A list of loaded rows.
     * @throws IOException Error accessing hbase.
     */
    private Collection<DataRow> next(long offset, int rowsNumber) throws IOException {
        LOG.info("Scanning next {} rows with offset {}...", rowsNumber, offset);
        int itemsNumber = rowsNumber <= GlobalConfig.instance().getBatchSizeForRead() ? rowsNumber : GlobalConfig.instance().getBatchSizeForRead();

        if (this.forwardNavigateOnly) {
            // Remove the current marker to reduce the memory load in case backward navigation should not be supported.
            popMarker();
        }

        Query query = getQuery();
        LOG.debug("Query: {}", query);
        try (RowScanner rowScanner = getRowScanner(query)) {
            Collection<DataRow> rows = new LinkedList<>();
            Marker marker = loadRows(rowScanner, offset, rowsNumber, rows);
            if (!marker.isEmpty()) {
                this.markers.push(marker);
            }

            return rows;
        }
    }

    /**
     * Removes a marker from the stack.
     *
     * @return A marker.
     */
    private Marker popMarker() {
        if (!this.markers.isEmpty()) {
            return this.markers.pop();
        }
        return null;
    }

    /**
     * Peeks the marker from the stack.
     *
     * @return A marker.
     */
    private Marker peekMarker() {
        if (!this.markers.isEmpty()) {
            return this.markers.peek();
        }
        throw new RuntimeException("No more markers found");
    }

    @Override
    public void close() throws IOException {
        if (markers != null) {
            markers.clear();
        }
        if (current != null) {
            current.clear();
        }
        if (columnTypes != null) {
            columnTypes.clear();
        }
    }
    //endregion

    /**
     * Represents a marking point between the batches of loaded rows.
     * Each time a new batch of rows is loaded this class holds the last key to start loading the next batch.
     */
    private static class Marker {

        //region Variables
        /**
         * The last key loaded from the previous batch of rows.
         */
        private byte[] endKey;
        /**
         * The start key loaded from the previous batch of rows.
         */
        private byte[] startKey;
        private int offset;
        private boolean hasNext;
        /**
         * A list of previously loaded rows.
         */
        private Collection<DataRow> rows;
        /**
         * A list of columns loaded from the rows.
         */
        private Collection<TypedColumn> columns;
        //endregion

        //region Constructor

        /**
         * Initializes a new instance of the {@link Marker} class.
         *
         * @param startKey The last loaded  start key.
         * @param endKey   The last loaded  end key.
         * @param hasNext  True when has next row.
         */
        private Marker(byte[] startKey, byte[] endKey, int offset, boolean hasNext, Collection<DataRow> rows, Collection<TypedColumn> columns) {
            this.startKey = startKey;
            this.endKey = endKey;
            this.offset = offset;
            this.hasNext = hasNext;
            this.rows = rows;
            this.columns = columns;
        }
        //endregion

        //region Public Methods
        public boolean isEmpty() {
            return startKey == null;
        }

        @Override
        public boolean equals(Object obj) {
            return obj instanceof Marker &&
                Arrays.equals(((Marker) obj).endKey, this.endKey) &&
                Arrays.equals(((Marker) obj).startKey, this.startKey);
        }

        @Override
        public int hashCode() {
            return (Bytes.toStringBinary(startKey) + Bytes.toStringBinary(endKey)).hashCode();
        }
        //endregion
    }
}
